package com.atguigu.realtime.app.dws;

import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.MyRedisUtil;
import com.atguigu.realtime.util.MyThreadPollUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/4/26 15:10
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {
    
    private Connection conn;
    private ThreadPoolExecutor pool;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(Constant.PHOENIX_DRIVER);
        conn = DriverManager.getConnection(Constant.PHOENIX_URL);
        
        pool = MyThreadPollUtil.getThreadPool();
    }
    
    protected abstract void addDim(T input, Connection conn, Jedis client);
    
    
    
    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) throws Exception {
        pool.execute(new Runnable() {
            @Override
            public void run() {
                Jedis client = MyRedisUtil.getClient();
                addDim(input, conn, client);
                resultFuture.complete(Collections.singletonList(input));
                client.close();
            }
        });
    }
    
    @Override
    public void close() throws Exception {
        if (conn != null) {
            conn.close();
        }
    
        if (pool != null) {
            pool.shutdown();
        }
    }
}
